package com.xhwl.data.mq.producer;

import com.alibaba.fastjson.JSONObject;
import com.xhwl.common.constant.cent.TopicConstant;
import com.xhwl.common.pojo.cent.sdata.personnel.StaffDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Nonnull;
import java.util.Objects;

/**
 * @author pengjunming
 * @description
 * @date 2022/7/5 13:50
 */
@Component
@Slf4j
public class StaffProducer {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    //员工离职主题
    private static String DELETE_TOPIC="cent-data-staff-delete";

    public void create(StaffDTO staff) {
        if (Objects.nonNull(staff)) {
            String content = JSONObject.toJSONString(staff);
            log.info("员工创建消息:{}", content);
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TopicConstant.CREATE_STAFF_TOPIC, content);
            future.addCallback(listener(content, TopicConstant.CREATE_STAFF_TOPIC));
        }
    }

    public void update(StaffDTO staff) {
        if (Objects.nonNull(staff)) {
            String content = JSONObject.toJSONString(staff);
            log.info("员工修改消息:{}", content);
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TopicConstant.UPDATE_STAFF_TOPIC, content);
            future.addCallback(listener(content, TopicConstant.UPDATE_STAFF_TOPIC));
        }
    }

    public void delete(StaffDTO staffDTO){
        if (Objects.nonNull(staffDTO)){
            String content = JSONObject.toJSONString(staffDTO);
            log.info("员工离职消息:{}", content);
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(DELETE_TOPIC, content);
            future.addCallback(listener(content, DELETE_TOPIC));
        }
    }

    private ListenableFutureCallback<SendResult<String, Object>> listener(String content, String topic) {

        return new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("推送成功 -主题:{} 内容：{}", topic, content);
            }

            @Override
            public void onFailure(@Nonnull Throwable ex) {
                log.info("推送失败 -主题:{} 内容：{}，信息：{}", topic, content, ex.getMessage());
            }
        };
    }
}
